import time

from .common_func.print_logger import PrintLogger as Logger

import_error_warned = False
try:
    import numpy as np
    import pandas as pd
    from joblib import Parallel, delayed
except ModuleNotFoundError:
    if not import_error_warned:
        Logger.warning("UStaticAlg is not valid, because require module:[numpy, pandas, joblib] is not installed. "
                       "If you want use UStaticAlg, please install [numpy, pandas, joblib] first")
        import_error_warned = True

class UStaticAlg:
    def __init__(self, logger=Logger, filter_abnormal = 1, filter_factor = 2, min_data_amount = 0):
        self._logger = logger
        self.filter_abnormal = filter_abnormal
        self.filter_factor = filter_factor
        self.min_data_amount = min_data_amount
        is_filter_str = filter_abnormal and '过滤异常数据' or '不过滤异常数据'
        min_data_str = min_data_amount and f"基线计算最小数据量要求:{min_data_amount}" or '基线计算无最小数据量要求'
        self._logger.info(f"统计算法类初始化完成，{is_filter_str}，{min_data_str} ")

    def parallel_group_egg(self, static_df: pd.DataFrame, group_ids: list, feature_field, feature_static_method):
        """
        使用joblib库的并行计算框架调动所有CPU核并行分组统计各组总数或唯一值总数
        :param static_df: 统计数据框
        :param group_ids: 特征分组字段集
        :param feature_field: 特征统计字段
        :param feature_static_method: 统计方法，支持 count:总数 和 nunique:唯一值数量
        :return:
        """
        self._logger.debug(f"[UStaticAlg.parallel_group_egg]::统计数据框共有{len(static_df)}行数据, "
                           f"分组聚合字段:{group_ids}, 特征字段:{feature_field}, "
                           f"统计方法:{feature_static_method}")
        _s = time.time()
        _group_res = static_df.groupby(group_ids)
        res = pd.DataFrame(Parallel(n_jobs=-1)(delayed(self.group_egg)(
            group_value, value_pd, feature_static_method, group_ids, feature_field) for group_value, value_pd in
                                               _group_res))
        self._logger.debug(f"[UStaticAlg.parallel_group_egg]执行完成，耗时:{round(time.time() - _s, 3)}秒")
        return res
    
    def parallel_dist_static(self, static_df: pd.DataFrame, group_ids: list, threshold_value: float):
        """
        使用joblib库的并行计算框架调动所有CPU核并行统计分布情况，具体算法见 dist_static
        例如：
        入参统计数据框为：
             group_id   feature_field   day1    day2
             1          5               2       3
             1          6               1       2
             1          7               1       8
        求和后统计数据框为：
             group_id   feature_field   day1    day2    COUNT
             1          5               2       3       5
             1          6               1       2       3
             1          7               1       8       9
        如果阈值为0.5，即50%，则经过 dist_static 计算后的统计结果为：
            group_id   baseline_value
                 1          [7]
        分布列表中只有7,因为5和6加在一起也没有超过50%
        :param static_df:
        :param group_ids:
        :param threshold_value: 阈值
        :return:
        """
        _s = time.time()
        # 获取统计数据列开始位置
        count_res_columns_start_pos = len(group_ids)
        # 从增加了feature_field的group_ids中再pop出特征统计字段名
        feature_field = group_ids.pop()
        self._logger.debug(f"[UStaticAlg.parallel_dist_static]::统计数据框共有{len(static_df)}行数据, "
                           f"分组聚合字段:{group_ids}, 特征字段:{feature_field}, "
                           f"分布范围阈值:{round(threshold_value * 100, 2)}%")
        # 对每一行的所有统计数据列求和
        static_df["COUNT"] = static_df.iloc[:, count_res_columns_start_pos:].sum(axis=1)

        # 从数据框中过滤掉过程统计数据列，只保留特征分组字段、特征统计字段和COUNT字段
        static_df = static_df.loc[:, group_ids + [feature_field, "COUNT"]]
        # 分组聚合
        _group_res = static_df.groupby(group_ids)
        self._logger.debug(f"[UStaticAlg.parallel_dist_static]::分组聚合后，共分为{len(_group_res)}组数据")
        # 并行计算出各行数据分布列表
        res = pd.DataFrame(Parallel(n_jobs=-1)(delayed(self.dist_static)(
            group_value, value_pd, threshold_value, group_ids, feature_field) for group_value, value_pd in _group_res))
        self._logger.debug(f"[UStaticAlg.parallel_dist_static]::分组的分布范围统计完成，获得{len(res)}行:分组分布范围数据")
        self._logger.debug(f"[UStaticAlg.parallel_dist_static]执行完成，耗时:{round(time.time() - _s, 3)}秒")
        return res
    
    def single_day_dist_static(self, static_df: pd.DataFrame, group_ids: list, feature_field: str):
        """
        单独统计具体某一天的分布情况，复用 dist_static 方法，将阈值置为1, 仅过滤值为0的空数据
        例如：
        入参统计数据框为：
             group_id   feature_field   2024-01-01
             1          5               0
             1          6               1
             1          7               2

        阈值固定设置为1，经过 dist_static 计算后的统计结果为：
            group_id   baseline_value
                 1          [6, 7]
        :param static_df:
        :param group_ids:
        :param feature_field:
        :return:
        """
        _s = time.time()
        columns_list = static_df.columns
        _date_time = columns_list[-1]
        self._logger.debug(f"[UStaticAlg.single_day_dist_static]::统计数据框共有{len(static_df)}行数据, "
                           f"分组聚合字段:{group_ids}, 特征字段:{feature_field}, 日期：{_date_time}")
        static_df = static_df.rename(columns={_date_time: "COUNT"})
        _group_res = static_df.groupby(group_ids)
        self._logger.debug(f"[UStaticAlg.single_day_dist_static]::分组聚合后，共分为{len(_group_res)}组数据")
        threshold_value = 1
        res = pd.DataFrame(Parallel(n_jobs=-1)(
            delayed(self.dist_static)(group_value, value_pd, threshold_value, group_ids, feature_field) for
            group_value, value_pd in _group_res))
        res = res.rename(columns={"baseline_value": _date_time})
        self._logger.debug(
            f"[UStaticAlg.single_day_dist_static]::分组的分布范围统计完成，获得{len(res)}行:分组分布范围数据")
        self._logger.debug(f"[UStaticAlg.single_day_dist_static]执行完成，耗时:{round(time.time() - _s, 3)}秒")
        return res

    def parallel_calc(self, static_df: pd.DataFrame, alg: str):
        """
        多进程并行计算
        输入为统计过程数据框中的数据列，输出为基线值Series
        并行计算的具体算法根据alg的值调用相应的类方法，如alg =std1，则调用self.std1来进行运算
        例如：
        入参统计数据框为：
             day1       day2    day3    day4    day5
             1          2       3       4       5
             2          3       4       5       6
             3          4       5       6       7

        算法为average，经过并行调用self.average 计算后的结果为一个Series：
                Series
                 3
                 4
                 5
        :param static_df:
        :param alg:
        :return:
        """
        self._logger.debug(f"[UStaticAlg.parallel_calc]::统计数据框共有{len(static_df)}行数据, 统计算法:{alg}")
        _s = time.time()
        calc_func = getattr(self, alg)
        calc_df = static_df.select_dtypes(include=np.number)
        res = pd.Series(Parallel(n_jobs=-1)(delayed(calc_func)(calc_df.iloc[i]) for i in range(len(calc_df))))
        self._logger.debug(
            f"[UStaticAlg.parallel_calc]::并行基线计算完成，获得{len(res)}行:基线数据")
        self._logger.debug(f"[UStaticAlg.parallel_calc]执行完成，耗时:{round(time.time() - _s, 3)}秒")
        return res

    def calc(self, static_df: pd.DataFrame, alg: str, const_value=0):
        """
        单进程计算，效率低，暂时不用
        :param static_df:
        :param alg:
        :param const_value:
        :return:
        """
        if alg == "const" and const_value:
            res_list = [const_value for _ in range(len(static_df))]
            return pd.Series(res_list)
        calc_func = getattr(self, alg)
        res = static_df.select_dtypes(include=np.number).apply(calc_func, axis=1)
        return res

    def abnormal_filter(self, row):
        """
        异常数据过滤，如果全局变量设置不过滤，则直接返回，否则根据过滤条件过滤掉异常数据，返回该行中的正常数据部分
        过滤逻辑，将一行数据进行计算，求出平均值和标准差，根据全局变量 OUTLIER_FACTOR 设置计算因子
        过滤掉在 [平均值-因子×标准差，平均值+因子×标准差]范围外的数据
        :param row:
        :return:
        """
        if not self.filter_abnormal:
            return row
        mean = row.mean()
        if len(row) == 1:
            std = 0
        else:
            std = row.std()

        return row[row.apply(lambda x: mean - self.filter_factor * std <= x <= mean + self.filter_factor * std)]

    def std1(self, row):
        """
        一倍标准差算法
        算法逻辑
         - 异常数据过滤
         - 求数据行的平均值和标准差
         - 以平均值 + 1 × 标准差 作为基线值返回
        :param row:
        :return:
        """
        row = self.abnormal_filter(row)
        if len(row) < self.min_data_amount:
            return pd.NaT
        _mean = row.mean()
        if len(row) == 1:
            _std = 0
        else:
            _std = row.std()
        result = _mean + _std
        if not result:
            return pd.NaT
        return result

    def std2(self, row):
        """
        2倍标准差算法
        算法逻辑
         - 异常数据过滤
         - 求数据行的平均值和标准差
         - 以平均值 + 2 × 标准差 作为基线值返回
        :param row:
        :return:
        """
        row = self.abnormal_filter(row)
        if len(row) < self.min_data_amount:
            return pd.NaT
        _mean = row.mean()
        if len(row) == 1:
            _std = 0
        else:
            _std = row.std()
        result = _mean + _std * 2
        if not result:
            return pd.NaT
        return result

    def std3(self, row):
        """
        3倍标准差算法
        算法逻辑
         - 异常数据过滤
         - 求数据行的平均值和标准差
         - 以平均值 + 3 × 标准差 作为基线值返回
        :param row:
        :return:
        """
        row = self.abnormal_filter(row)
        if len(row) < self.min_data_amount:
            return pd.NaT
        _mean = row.mean()
        if len(row) == 1:
            _std = 0
        else:
            _std = row.std()
        result = _mean + _std * 3
        if not result:
            return pd.NaT
        return result

    def median(self, row):
        """
        中位数算法
        算法逻辑
         - 异常数据过滤
         - 求数据行的中位数
         - 以 中位数 作为基线值返回
        :param row:
        :return:
        """
        row = self.abnormal_filter(row)
        if len(row) < self.min_data_amount:
            return pd.NaT
        result = row.median()
        if not result:
            return pd.NaT
        return result

    def average(self, row):
        """
        中位数算法
        算法逻辑
         - 异常数据过滤
         - 求数据行的中位数
         - 以 中位数 作为基线值返回
        :param row:
        :return:
        """
        row = self.abnormal_filter(row)
        if len(row) < self.min_data_amount:
            return pd.NaT
        result = row.mean()
        if not result:
            return pd.NaT
        return result

    @staticmethod
    def dist_static(group_value, value_pd, threshold_value, group_ids, feature_field):
        """
        分布情况统计
        统计逻辑
            - 对分组数据框的COUNT列数据进行升序排列
            - 计算数据的总和
            - 计算累加和，作为一列放到数据框中
            - 计算每一行的累加合在总和中的占比
            - 过滤出占比大于（1-阈值）的数据行
        举例：
            入参数据框：
                group_id    feature_field   COUNT
                1           5               1
                1           6               10
                1           7               5
                1           8               20
            升序排列后：
                group_id    feature_field   COUNT
                1           5               1
                1           7               5
                1           6               10
                1           8               20
            求和结果为36
            累加和计算后：
                group_id    feature_field   COUNT   CUM_SUM
                1           5               1       1
                1           7               5       6
                1           6               10      16
                1           8               20      36
            计算后，（最后一列不存在，直接作为过滤条件，列出来是方便理解）：
                group_id    feature_field   COUNT   CUM_SUM     累加和除以总和
                1           5               1       1           2.78%
                1           7               5       6           16.67%
                1           6               10      16          44.44%
                1           8               20      36          100%
            如果阈值设置为0.8,即80%，则过滤掉累加和/总和低于0.2（20%）的数据行，即过滤掉前两行
            返回结果为
                group_id    baseline_value
                1           [6,8]
        :param group_value: 分组聚合字段的值
        :param value_pd: 该分组的数据框
        :param threshold_value: 阈值，以浮点数表示
        :param group_ids: 分组聚合字段列表
        :param feature_field: 特征分布统计字段
        :return:
        """
        sorted_pd = value_pd.sort_values(by="COUNT", ascending=True)
        _sum = sorted_pd.COUNT.sum()
        sorted_pd["CUM_SUM"] = sorted_pd.COUNT.cumsum()
        baseline_value_df = sorted_pd[sorted_pd.CUM_SUM / _sum > 1 - threshold_value]
        res = dict(zip(group_ids, group_value))
        res["baseline_value"] = list(baseline_value_df[feature_field])
        if not res["baseline_value"]:
            res["baseline_value"] = pd.NaT
        return res

    
    @staticmethod
    def group_egg(group_value, value_pd, feature_static_method, group_ids, feature_field):
        """
        对分组数据进行统计处理，支持唯一值数量nunique和总数count两种统计方法
        举例：
            入参数据框：
                group_id    feature_field
                1           5
                1           6
                1           7
                1           8
                1           7
                1           8
                1           7
                1           8
            nunique统计结果为
                group_id    feature_field
                    1           4
            count统计结果为
                group_id    feature_field
                    1           8
        :param group_value:
        :param value_pd:
        :param feature_static_method:
        :param group_ids:
        :param feature_field:
        :return:
        """
        # 动态获取统计方法并实施统计， 比如当feature_static_method为nunique时，feature_field为api_id时，
        # 以下两行代码动态执行结果等同于 _group_res = value_pd["api_id"].nunique()
        _method = getattr(value_pd[feature_field], feature_static_method)
        _group_res = _method()
        # 将分组字段名和字段值组成字典，并将统计结果更新进去，以feature_field为key
        if not isinstance(group_value, (list, tuple)):
            group_value = [group_value]
        res = dict(zip(group_ids, group_value))
        res[feature_field] = _group_res
        return res
